package com.tianhai.db;

import java.io.IOException;
import java.io.RandomAccessFile;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.locks.ReentrantReadWriteLock;

/**
 * WAL预写日志 - Write-Ahead Logging
 * 实现事务日志记录和崩溃恢复
 */
public class WAL {
    
    private final String walFilePath;
    private FileChannel fileChannel;
    private RandomAccessFile randomAccessFile;
    private final ReentrantReadWriteLock rwLock;
    private long nextLogOffset;
    
    // 日志记录类型
    public enum LogType {
        BEGIN(1),
        COMMIT(2),
        ABORT(3),
        INSERT(4),
        UPDATE(5),
        DELETE(6);
        
        private final int code;
        
        LogType(int code) {
            this.code = code;
        }
        
        public int getCode() {
            return code;
        }
        
        public static LogType fromCode(int code) {
            for (LogType type : values()) {
                if (type.code == code) return type;
            }
            throw new IllegalArgumentException("未知日志类型: " + code);
        }
    }
    
    /**
     * 构造函数
     */
    public WAL(String walFilePath) throws IOException {
        this.walFilePath = walFilePath;
        this.rwLock = new ReentrantReadWriteLock();
        initializeWAL();
    }
    
    /**
     * 初始化WAL文件
     */
    private void initializeWAL() throws IOException {
        boolean fileExists = Files.exists(Paths.get(walFilePath));
        
        this.randomAccessFile = new RandomAccessFile(walFilePath, "rw");
        this.fileChannel = randomAccessFile.getChannel();
        
        if (fileExists) {
            // 文件存在，定位到末尾
            this.nextLogOffset = fileChannel.size();
        } else {
            // 新文件，从0开始
            this.nextLogOffset = 0;
        }
    }
    
    /**
     * 记录事务日志
     */
    public void logTransaction(long txnId, LogType logType, String tableName, String data) throws IOException {
        rwLock.writeLock().lock();
        try {
            // 构建日志记录
            LogRecord record = new LogRecord(txnId, logType, tableName, data);
            byte[] recordBytes = record.serialize();
            
            // 写入文件
            ByteBuffer buffer = ByteBuffer.wrap(recordBytes);
            fileChannel.write(buffer, nextLogOffset);
            
            // 强制刷新到磁盘（确保持久化）
            fileChannel.force(true);
            
            // 更新偏移量
            nextLogOffset += recordBytes.length;
            
        } finally {
            rwLock.writeLock().unlock();
        }
    }
    
    /**
     * 恢复事务
     * 返回已提交的事务列表
     */
    public List<Transaction> recover() throws IOException {
        rwLock.readLock().lock();
        try {
            List<Transaction> transactions = new ArrayList<>();
            
            if (nextLogOffset == 0) {
                return transactions; // 无日志记录
            }
            
            // 从头读取所有日志
            long offset = 0;
            while (offset < nextLogOffset) {
                LogRecord record = readLogRecord(offset);
                if (record == null) break;
                
                // 查找或创建事务
                Transaction txn = findOrCreateTransaction(transactions, record.txnId);
                txn.addLogRecord(record);
                
                // 根据日志类型更新事务状态
                switch (record.logType) {
                    case BEGIN:
                        txn.status = TransactionStatus.ACTIVE;
                        break;
                    case COMMIT:
                        txn.status = TransactionStatus.COMMITTED;
                        break;
                    case ABORT:
                        txn.status = TransactionStatus.ABORTED;
                        break;
                    default:
                        // INSERT/UPDATE/DELETE操作
                        break;
                }
                
                offset += record.getSerializedSize();
            }
            
            // 只返回已提交的事务
            List<Transaction> committedTxns = new ArrayList<>();
            for (Transaction txn : transactions) {
                if (txn.status == TransactionStatus.COMMITTED) {
                    committedTxns.add(txn);
                }
            }
            
            return committedTxns;
            
        } finally {
            rwLock.readLock().unlock();
        }
    }
    
    /**
     * 读取日志记录
     */
    private LogRecord readLogRecord(long offset) throws IOException {
        try {
            // 读取记录大小（4字节）
            ByteBuffer sizeBuffer = ByteBuffer.allocate(4);
            int bytesRead = fileChannel.read(sizeBuffer, offset);
            if (bytesRead < 4) return null;
            
            sizeBuffer.flip();
            int recordSize = sizeBuffer.getInt();
            
            // 读取完整记录
            ByteBuffer recordBuffer = ByteBuffer.allocate(recordSize);
            fileChannel.read(recordBuffer, offset);
            recordBuffer.flip();
            
            return LogRecord.deserialize(recordBuffer.array());
            
        } catch (Exception e) {
            return null;
        }
    }
    
    /**
     * 查找或创建事务
     */
    private Transaction findOrCreateTransaction(List<Transaction> transactions, long txnId) {
        for (Transaction txn : transactions) {
            if (txn.txnId == txnId) {
                return txn;
            }
        }
        Transaction newTxn = new Transaction(txnId);
        transactions.add(newTxn);
        return newTxn;
    }
    
    /**
     * 清空WAL日志
     */
    public void truncate() throws IOException {
        rwLock.writeLock().lock();
        try {
            fileChannel.truncate(0);
            nextLogOffset = 0;
        } finally {
            rwLock.writeLock().unlock();
        }
    }
    
    /**
     * 关闭WAL
     */
    public void close() throws IOException {
        rwLock.writeLock().lock();
        try {
            if (fileChannel != null && fileChannel.isOpen()) {
                fileChannel.force(true);
                fileChannel.close();
            }
            if (randomAccessFile != null) {
                randomAccessFile.close();
            }
        } finally {
            rwLock.writeLock().unlock();
        }
    }
    
    /**
     * 日志记录
     */
    public static class LogRecord {
        public final long txnId;
        public final LogType logType;
        public final String tableName;
        public final String data;
        
        public LogRecord(long txnId, LogType logType, String tableName, String data) {
            this.txnId = txnId;
            this.logType = logType;
            this.tableName = tableName != null ? tableName : "";
            this.data = data != null ? data : "";
        }
        
        /**
         * 序列化日志记录
         * 格式: [总大小(4)] [txnId(8)] [logType(4)] [tableNameLen(4)] [tableName] [dataLen(4)] [data]
         */
        public byte[] serialize() {
            byte[] tableNameBytes = tableName.getBytes(StandardCharsets.UTF_8);
            byte[] dataBytes = data.getBytes(StandardCharsets.UTF_8);
            
            int totalSize = 4 + 8 + 4 + 4 + tableNameBytes.length + 4 + dataBytes.length;
            ByteBuffer buffer = ByteBuffer.allocate(totalSize);
            
            buffer.putInt(totalSize);              // 总大小
            buffer.putLong(txnId);                 // 事务ID
            buffer.putInt(logType.getCode());      // 日志类型
            buffer.putInt(tableNameBytes.length);  // 表名长度
            buffer.put(tableNameBytes);            // 表名
            buffer.putInt(dataBytes.length);       // 数据长度
            buffer.put(dataBytes);                 // 数据
            
            return buffer.array();
        }
        
        /**
         * 反序列化日志记录
         */
        public static LogRecord deserialize(byte[] bytes) {
            ByteBuffer buffer = ByteBuffer.wrap(bytes);
            
            int totalSize = buffer.getInt();
            long txnId = buffer.getLong();
            LogType logType = LogType.fromCode(buffer.getInt());
            
            int tableNameLen = buffer.getInt();
            byte[] tableNameBytes = new byte[tableNameLen];
            buffer.get(tableNameBytes);
            String tableName = new String(tableNameBytes, StandardCharsets.UTF_8);
            
            int dataLen = buffer.getInt();
            byte[] dataBytes = new byte[dataLen];
            buffer.get(dataBytes);
            String data = new String(dataBytes, StandardCharsets.UTF_8);
            
            return new LogRecord(txnId, logType, tableName, data);
        }
        
        /**
         * 获取序列化后的大小
         */
        public int getSerializedSize() {
            return 4 + 8 + 4 + 4 + tableName.getBytes(StandardCharsets.UTF_8).length + 
                   4 + data.getBytes(StandardCharsets.UTF_8).length;
        }
        
        @Override
        public String toString() {
            return "LogRecord{txnId=" + txnId + ", type=" + logType + 
                   ", table=" + tableName + ", data=" + data + "}";
        }
    }
    
    /**
     * 事务状态
     */
    public enum TransactionStatus {
        ACTIVE,
        COMMITTED,
        ABORTED
    }
    
    /**
     * 事务
     */
    public static class Transaction {
        public final long txnId;
        public TransactionStatus status;
        public final List<LogRecord> logRecords;
        
        public Transaction(long txnId) {
            this.txnId = txnId;
            this.status = TransactionStatus.ACTIVE;
            this.logRecords = new ArrayList<>();
        }
        
        public void addLogRecord(LogRecord record) {
            logRecords.add(record);
        }
        
        @Override
        public String toString() {
            return "Transaction{txnId=" + txnId + ", status=" + status + 
                   ", logs=" + logRecords.size() + "}";
        }
    }
}
